package cn.witsky.rtp2ws.push.retry;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.witsky.rtp2ws.cache.CallManager;
import cn.witsky.rtp2ws.constant.ThreadPoolConstant;
import cn.witsky.rtp2ws.domain.CallInfo;
import cn.witsky.rtp2ws.domain.Rtp;
import cn.witsky.rtp2ws.util.CapcityUtil;
import cn.witsky.rtp2ws.util.DefaultThreadFactory;
import cn.witsky.rtp2ws.util.PushUtil;
import cn.witsky.rtp2ws.util.RedisUtil;

public class RetryManager {
	private static Logger logger = LoggerFactory.getLogger(RetryManager.class);
	private static ThreadPoolExecutor readPool = null;
	private static ThreadPoolExecutor downLoadPool = null;

	public static int keepAliveTime = 60;
	public static TimeUnit timeUnit = TimeUnit.SECONDS;
	public static RejectedExecutionHandler handler = new DiscardPolicy();
	public static Thread checkTimeOutThread = null;
	private static Map<String, AtomicInteger> callRtpMap = new ConcurrentHashMap<String, AtomicInteger>(1024);
	private static List<String> freeCallIdList = new CopyOnWriteArrayList<String>();

	public static void freeRtp(String callId) {
		AtomicInteger rtpCount = callRtpMap.get(callId);
		if (rtpCount.decrementAndGet() <= 0) {
			freeCallIdList.add(callId);
		}
	}

	public static void initialize() {
		initReadPool();
		initDownLoadPool();
		checkTimeOutThread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						Iterator<String> callIds = freeCallIdList.iterator();
						while (callIds.hasNext()) {
							String callId = callIds.next();
							CallInfo call = CallManager.getCall(callId);
							if(call==null) {
								callRtpMap.remove(callId);
							}else {
								Rtp caller=call.getRtp(true);
								Rtp called=call.getRtp(false);
								if(caller.getTaskStatus()&&called.getTaskStatus()) {
									AtomicInteger rtpCount = callRtpMap.get(callId);
									if (rtpCount.decrementAndGet() <= 0) {
										if(caller.getTaskStatus()&&called.getTaskStatus()) {
											callRtpMap.remove(callId);
											CallManager.removeCall(callId);
											PushUtil.stop(call);
											freeCallIdList.remove(callId);
											logger.info("执行清理{}",callId);
										}
									}
								}
							}
						}
						TimeUnit.MILLISECONDS.sleep(500);
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

				}
			}
		});
		checkTimeOutThread.setName("rtp2ws-retry-CheckTimeOut-Thread");
		checkTimeOutThread.start();

	}

	private static void initReadPool() {
		int corePoolSize = 10;
		int maximumPoolSize = 20;
		int times = CapcityUtil.getTime();
		corePoolSize = corePoolSize * times;
		if (corePoolSize > 100) {
			corePoolSize = 100;
		}
		maximumPoolSize = maximumPoolSize * times;
		if (maximumPoolSize > 200) {
			maximumPoolSize = 200;
		}
		BlockingQueue<Runnable> waitQueue = new LinkedBlockingQueue<Runnable>(2000000);
		readPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, waitQueue,new DefaultThreadFactory(ThreadPoolConstant.retryRead), handler);
		logger.info("初始化{}线程池 corePoolSize:{} maximumPoolSize:{} waitQueueSzie:{} keepAliveTime:{} timeUnit:{}",ThreadPoolConstant.retryRead, corePoolSize, maximumPoolSize, 2000000, keepAliveTime, timeUnit);
	}

	private static void initDownLoadPool() {
		int corePoolSize = 10;
		int maximumPoolSize = 20;
		int times = CapcityUtil.getTime();
		corePoolSize = corePoolSize * times;
		if (corePoolSize > 50) {
			corePoolSize = 50;
		}
		maximumPoolSize = maximumPoolSize * times;
		if (maximumPoolSize > 100) {
			maximumPoolSize = 100;
		}
		BlockingQueue<Runnable> waitQueue = new LinkedBlockingQueue<Runnable>(1000000);
		downLoadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, waitQueue,new DefaultThreadFactory(ThreadPoolConstant.retryDownload), handler);
		logger.info("初始化{}线程池 corePoolSize:{} maximumPoolSize:{} waitQueueSzie:{} keepAliveTime:{} timeUnit:{}",ThreadPoolConstant.retryDownload, corePoolSize, maximumPoolSize, 1000000, keepAliveTime, timeUnit);
	}

	public static boolean offerCall(String callId) {
		boolean isSuc = false;
		if (callRtpMap.containsKey(callId)) {
			logger.warn("call:{}任务已经在执行了",callId);
			return isSuc;
		}
		if (CallManager.hasCall(callId)) {
			logger.warn("call:{}会话已经存在",callId);
			return isSuc;
		}
		CallInfo call = RedisUtil.getBackUpCallInfo(callId);
		if (call != null) {
			CallManager.addCalls(call, false);
			AtomicInteger rtpCount = new AtomicInteger();
			callRtpMap.put(callId, rtpCount);
			execDownLoadTask(new DownLoadTask(call));
			isSuc = true;
		}
		return isSuc;
	}

	public static AtomicInteger getRtpCount(String callId) {
		return callRtpMap.get(callId);
	}

	public static void execReadTask(ReadTask task) {
		readPool.execute(task);
	}

	public static void execDownLoadTask(DownLoadTask task) {
		downLoadPool.execute(task);
	}
}
